【Java多线程】JUC集合 07. ConcurrentLinkedQueue

ConcurrentLinkedQueue

1. 前言

• 基于链表的线程安全的队列，适用于高并发
• 无界FIFO队列

2. 源码分析

2.1 数据结构

public class ConcurrentLinkedQueue<E> extends AbstractQueue<E>
        implements Queue<E>, java.io.Serializable {
    
    private transient volatile Node<E> head;

    private transient volatile Node<E> tail;
}

/**
*Node内部类
*/
private static class Node<E> {
    volatile E item;
    volatile Node<E> next;

    /**
    * Constructs a new node.  Uses relaxed write because item can
    * only be seen after publication via casNext.
    */
    Node(E item) {
        UNSAFE.putObject(this, itemOffset, item);
    }
}

UML类图如下：

ConcurrentLinkedQueue的链表Node中的item和next的类型是volatile，通过CAS来设置值。

ConcurrentLinkedQueue是通过volatile + CAS来实现多线程对竞争资源的互斥访问的

2.2 核心函数

• 构造函数

public ConcurrentLinkedQueue() {
    head = tail = new Node<E>(null);
}

//创建一个最初包含给定 collection 元素的 ConcurrentLinkedQueue，按照此 collection 迭代器的遍历顺序来添加元素。
public ConcurrentLinkedQueue(Collection<? extends E> c) {
    Node<E> h = null, t = null;
    for (E e : c) {
        checkNotNull(e);
        Node<E> newNode = new Node<E>(e);
        if (h == null)
            h = t = newNode;
        else {
            t.lazySetNext(newNode);
            t = newNode;
        }
    }
    if (h == null)
        h = t = new Node<E>(null);
    head = h;
    tail = t;
}

• 添加元素: offer(E e)

public boolean offer(E e) {
    checkNotNull(e);//不允许空
    final Node<E> newNode = new Node<E>(e);

    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        if (q == null) {
            // p is last node
            if (p.casNext(null, newNode)) {//q是尾节点p的next,设置p的下一节点为newNode
                // Successful CAS is the linearization point
                // for e to become an element of this queue,
                // and for newNode to become "live".
                if (p != t) // hop two nodes at a time
                    casTail(t, newNode);  // Failure is OK.
                return true;
            }
            // Lost CAS race to another thread; re-read next
        }
        else if (p == q)
            // We have fallen off list.  If tail is unchanged, it
            // will also be off-list, in which case we need to
            // jump to head, from which all live nodes are always
            // reachable.  Else the new tail is a better bet.
            p = (t != (t = tail)) ? t : head;//若尾节点没有发生变化的话，那么，应该是头节点发生了变化，则设置p为头节点，然后重新遍历链表；否则(尾节点变化的话)，则设置p为尾节点。
        else
            // Check for tail updates after two hops.
            p = (p != t && t != (t = tail)) ? t : q;//如果p和t相等，则设置p为q。否则的话，判断“尾节点是否发生变化”，没有变化的话，则设置p为q；否则，设置p为尾节点
    }
}

• 取出元素: poll()

public E poll() {
    // 设置“标记”
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            E item = p.item;

            // 情况1
            // 表头的数据不为null，并且“设置表头的数据为null”这个操作成功的话;
            // 则比较“p和h”(若p!=h，即表头发生了变化，则更新表头，即设置表头为p)，然后返回原表头的item值。
            if (item != null && p.casItem(item, null)) {
                if (p != h) // hop two nodes at a time
                    updateHead(h, ((q = p.next) != null) ? q : p);
                return item;
            }
            // 情况2
            // 表头的下一个节点为null，即链表只有一个“内容为null的表头节点”。则更新表头为p，并返回null。
            else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            }
            // 情况3
            // 这可能到由于“情况4”的发生导致p=q，在该情况下跳转到restartFromHead标记重新操作。
            else if (p == q)
                continue restartFromHead;
            // 情况4
            // 设置p为q
            else
                p = q;
        }
    }
}

//更新表头为p，并且设置h的next为自身，帮助gc
final void updateHead(Node<E> h, Node<E> p) {
    if (h != p && casHead(h, p))
        h.lazySetNext(h);
}

• size()

//遍历链表获取size
public int size() {
    int count = 0;
    for (Node<E> p = first(); p != null; p = succ(p))
        if (p.item != null)
            // Collection.size() spec says to max out
            if (++count == Integer.MAX_VALUE)
                break;
    return count;
}
//返回队列头节点
Node<E> first() {
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            boolean hasItem = (p.item != null);
            if (hasItem || (q = p.next) == null) {
                updateHead(h, p);
                return hasItem ? p : null;
            }
            else if (p == q)
                continue restartFromHead;
            else
                p = q;
        }
    }
}

3. 参考

http://www.cnblogs.com/skywang12345/p/3498995.html